package crawler.douban;

import com.alibaba.fastjson.JSON;
import com.mongodb.client.MongoCollection;
import crawler.http.HttpClientTest;
import crawler.proxy.Proxy;
import crawler.proxy.uitl.ProxyList;
import mongo.MongoTest;
import mongo.SpringMongo;
import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentSkipListSet;

/**
 * https://movie.douban.com/subject/11600078/
 * 豆瓣爬虫
 * USER: lintc 【lintiancong@zhuojianchina.com】
 * DATE: 2017-09-08 10:11
 */
public class DouBanSpider {
    MongoCollection<org.bson.Document> collection;
    // 解析错误集合
    MongoCollection<org.bson.Document> errorCollection;
    // 抓取线程数
    private int catchPoolCount;
    // 抓取数量
    private int catchCount;
    // 网页编码
    private String charset;
    // 待抓取
    public static Queue<Movie> waitList = new PriorityQueue<>((o1, o2) -> (int) ((o2.getStartLevel() - o1.getStartLevel()) * 10));
    // 正在解析
    public static Set<String> doingList = new ConcurrentSkipListSet<>();
    // 已抓取
    public static Set<String> finishList = new ConcurrentSkipListSet<>();
    // 已获取的全部列表（防止重复抓取）
    public static Set<String> haveList = new ConcurrentSkipListSet<>();

    private List<DouBanRunnable> spiderList = new ArrayList<>();

    /**
     * 爬虫控制器
     * @param catchPoolCount 线程数
     * @param catchCount 爬取数
     * @param charset 网页编码
     * @param startUrl 开始抓取地址
     */
    public DouBanSpider(int catchPoolCount, int catchCount, String charset, String startUrl) {
        // 1 从数据库中获取已拥有的
        haveList = SpringMongo.getDbMovie();
        collection = MongoTest.test.getCollection("db_movie");
        errorCollection = MongoTest.test.getCollection("error_db_movie");
        this.catchCount = catchCount;
        this.catchPoolCount = catchPoolCount;
        this.charset = charset;
//        this.proxyList = new ProxyList(30);
//        new Thread(proxyList).start();
        ProxyList.initialization();
        // 初始化 待抓取池
        new Thread(new DBMovieListCatch(1)).start();
        System.out.println("********爬虫初始化成功********");
    }

    public void start() {
        for (int i = 0; i < catchPoolCount; i++) {
            spiderList.add(new DouBanRunnable());
        }
        for (DouBanRunnable douBanRunnable : spiderList) {
            new Thread(douBanRunnable).start();
        }
        while (finishList.size() < catchCount) { // 判断是否已抓取指定数量
            friendlyToDouban(null);
        }
        stop();
    }

    // 结束爬取
    public void stop() {
        ProxyList.proxyStop();
        for (DouBanRunnable douBanRunnable : spiderList) {
            douBanRunnable.stop();
        }
    }

    /**
     * 抓取电影
     * 返回 代理方便复用
     * @param movie
     */
    public Proxy catchMovie(Proxy proxy, Movie movie) {
        System.out.println("爬虫开始：" + movie.getUrl() + " (代理): " + ProxyList.proxyIpList.size());
        long time = System.currentTimeMillis();
        String content = HttpClientTest.proxy2(proxy, movie.getUrl(), charset);
        while (StringUtils.isBlank(content)) { // 验证proxy是否失效
            if ("404".equals(content)) { // 页面爬取失败
                System.out.println("爬取失败 (" + content + "): (地址)" + movie.getUrl());
                return proxy;
            }
            long timeCost = System.currentTimeMillis() - time;
            System.out.println(proxy + ": 失效 : （代理）" + ProxyList.proxyIpList.size() + " : （耗时）" + timeCost + ": (地址)" + movie.getUrl());
            time = System.currentTimeMillis();
            proxy = ProxyList.getProxy();
            content = HttpClientTest.proxy2(proxy, movie.getUrl(), charset);
        }
        // 1 抓取电影详情
        Document d = Jsoup.parse(content);
        try {
            movie.setHtml(d);
            String str = JSON.toJSONString(movie);
            MongoTest.insertOne(collection, JSONObject.fromObject(str));
            // 2 获取相似电影
            List<Movie> l = getList(d);
            for (Movie movie1 : l) {
                setWaitList(movie1);
            }
        } catch (Exception e) {
            System.out.println(movie.getUrl() + ": 解析失败 :" + e.getMessage());
            JSONObject obj = new JSONObject();
            if (movie != null) {
                String str = JSON.toJSONString(movie);
                obj = JSONObject.fromObject(str);
            }
            obj.put("url", movie.getUrl());
            MongoTest.insertOne(errorCollection, obj);
        }
        // 3 代理复用
//        ProxyList.proxyIpList.add(proxy);
        return proxy;
    }

    /**
     * 线程安全
     * 保存待抓取
     * @param movie
     */
    public static synchronized void setWaitList(Movie movie) {
        if (!haveList.contains(movie.getUrl())) {
            haveList.add(movie.getUrl()); // 已获取的网址，防止重复抓取
            waitList.add(movie);
        }
    }

    /**
     * 获取 相似电影
     * @param doc
     * @return
     */
    public static List<Movie> getList(Document doc) {
        List<Movie> list = new ArrayList<>();
        Elements elements = doc.select(".recommendations-bd dd a");
        for (Element element : elements) {
            Movie movie = new Movie();
            movie.setUrl(element.attr("href"));
            movie.setName(element.text());
            list.add(movie);
        }
        return list;
    }

    /**
     * 获取 待抓取电影
     * @return
     */
    public Movie getCatchMovie() {
        Movie movie = waitList.poll();
        while (null == movie) {
            movie = waitList.poll();
            friendlyToDouban(1);
        }
        doingList.add(movie.getUrl());
        return movie;
    }

    //sleep for the random second so that avoiding to be listed into blacklist
    public static long friendlyToDouban(Integer time)  {
        if (null == time) {
            time = (new Random().nextInt(10) + 1);
        }
        try {
            Thread.sleep(time * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return time;
    }

    public static void test() {
        Proxy proxy = new Proxy();
        proxy.setIp("58.19.80.127");
        proxy.setPort("808");
        String content = HttpClientTest.proxy2(proxy, HttpClientTest.URL, "gb2312");
        System.out.println(content);
    }

    public static void main(String[] args) throws IOException {
        String url = "https://movie.douban.com/subject/1781026/?from=subject-page";
        DouBanSpider spider = new DouBanSpider(8, 1000, null, url);
        spider.start();
    }

    /**
     * 抓取线程
     */
    class DouBanRunnable implements Runnable {

        private boolean start = true;

        @Override
        public void run() {
            // 1 获取代理
            Proxy proxy = ProxyList.getProxy();
            while (start ||
                    (finishList.size() < catchCount && // 抓取数未达标
                    !(waitList.size() == 0 && finishList.size() > 0))) { // 抓取已开始，待抓取池已空 TODO 这种情况下会造成死循环
                Movie movie = getCatchMovie();
                String catchUrl = movie.getUrl();
                long time = 0;
                try {
                    // 2 获取电影信息
                    proxy = catchMovie(proxy, movie);
                } catch (Exception e) {
                    setWaitList(movie);
                    System.out.println(movie.getName() + " : " + movie.getUrl() + " : 抓取失败" + " :(待抓取) " + waitList.size() + " :(正执行) " + doingList.size());
                    continue;
                } finally {
                    doingList.remove(catchUrl);
                    time = friendlyToDouban(null);
                }
                // 3 记录已爬取
                finishList.add(catchUrl);
                System.out.println(finishList.size() + ":" + movie.getName() + " :(待抓取) " + waitList.size() + " :(正执行) " + doingList.size() + " : " + time);
            }
            System.out.println("抓取结束: (已抓取) " + finishList.size() + " :(未抓取) " + waitList.size());
        }

        public void stop() {
            start = false;
        }
    }
}
